package com.atguigu.edu.realtime.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.utils.DimUtil;
import com.atguigu.edu.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import java.util.Collections;
import java.util.concurrent.ExecutorService;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T> {
// 发送异步请求，完成维度关联的具体类
/*
* 模板方法设计模式
* 在父类中定义完成某一个核心算法骨架（步骤），具体的实现类可以延迟到子类中去完成
* 在不改变父类核心算法骨架的前提下，每一个子类都可以有自己不同的实现
*
* */
private ExecutorService executorService;
private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
       executorService  = ThreadPoolUtil.getInstance();
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
       //开启新的线程，发送异步请求
       executorService.submit(
               new Runnable() {
                   @Override
                   public void run() {
                       //根据流中的元素获取要关联的维度主键
                      String key =getKey(obj);
                       //根据维度主键获取维度对象
                       JSONObject dimInfoJsonObj = DimUtil.getDimInfo(tableName, key);
                       //将维度对象属性补充到流中的对象上
                       if(dimInfoJsonObj != null){
                           join(dimInfoJsonObj,obj);
                       }
                   //获取数据库交互的结果并发送给ResultFuture的回调函数
                       resultFuture.complete(Collections.singleton(obj));

                   }
               }
       );
    }
}
